#include <stdlib.h>
#include <iostream>
#include <iomanip>
#include <unistd.h>

/* DG: gethrtime() isn't easily available on Linux */
#ifdef sparc
#include <sys/time.h>
#endif

#include "fatals.h"
#include "MapReduceScheduler.h"

using namespace std;

/* Key Comparison Function -- Keys are int* */
int intcmp( const void *v1, const void * v2 ) {
  const int * i1 = (const int *) v1;
  const int * i2 = (const int *) v2;

  if( *i1 < *i2 )      return -1; 
  else if( *i1 > *i2 ) return  1;
  else                 return  0;
}

/* Map function */
void sumarray_map( map_args_t * args ) {
  int nChunkSize  = args->length; 
  // cout << "Map chunk size is " << nChunkSize << endl;
  int * miniArray = (int*) args->data;

  // cout << "Calculating intermediate sum over " << nChunkSize << " elements: " ;

  int intermediate_sum = 0;
  for(int i=0;i<nChunkSize;i++) {
    intermediate_sum += miniArray[i];
  }

  // cout << intermediate_sum << endl;

  int * key = new int;
  *key = 1;

  int * val = new int;
  *val = intermediate_sum;

  // cout << "Emitting intermediate <" << *key << "," << *val << ">" << endl;
  emit_intermediate( key, val, sizeof( int* ) );
}

/* Reduce Function */
void sumarray_reduce( void * key_in, void ** vals_in, int vals_len ) {
  int nElements = vals_len;
  int ** p_array = (int**) vals_in;
  int * p_key = (int*) key_in;

  delete p_key;

  int sum = 0;
  for(int i=0;i<nElements;i++) {
    sum += p_array[i][0];
    delete p_array[i];
  }

  int * key = new int;
  *key = 0;

  int * val = new int;
  *val = sum;

  emit( key, val );
}

int main( int argc, char * argv[] ) {

  scheduler_args_t sched_args;
  final_data_t result;
  int * array;
  int nElements;
#ifdef sparc
  hrtime_t starttime, endtime, linearsumtime, mr_sumtime;
#endif

  /* Parameter parsing and checking */
  if( argc != 2 ) {
    fatal("Usage:\n%s <arraySize>\n", argv[0]);
  }

  nElements = atoi( argv[1] );
  if( nElements <= 0 ) {
    fatal("%s is not a valid array size.\n", argv[1]);
  }

  array = new int[nElements];

  /* Initialize the array */
  for(int i=0;i<nElements;i++) {
    array[i] = i;
  }

#ifdef sparc
  starttime = gethrtime();
#endif
  int true_sum = 0;
  for(int i=0;i<nElements;i++) {
    true_sum += i;
  }
#ifdef sparc
  endtime = gethrtime();
  linearsumtime = endtime-starttime;
#endif

  /*---------------------------------------------------------+
   | Set up the arguments to the scheduler                   |
   +---------------------------------------------------------*/

  /* A pointer to whatever the splitter will split */
  sched_args.task_data = array;

  /* Total number of bytes of data */
  sched_args.data_size = nElements*sizeof(int);

  /* Pointer to the map function (REQUIRED) */
  sched_args.map = sumarray_map;

  /* Pointer to the reduce function ( NULL => Identity function ) */
  sched_args.reduce = sumarray_reduce;

  /* Pointer to the splitter function ( NULL => Array Splitter ) */
  sched_args.splitter = NULL; 

  /* Pointer to the Key Comparison function (REQUIRED) */
  sched_args.key_cmp = intcmp;

  /* Pointer to the final output data (allocated by USER, NOT by runtime) */
  sched_args.result = &result;

  /* Pointer to the partition function ( default is a hash ) */
  sched_args.partition = NULL;

  /*-----------------------------------------------------------+
   | Stuff below this point is SUPPOSEDLY only for performance |
   | tuning... in my experience, these can screw things up, so |
   | be wary.                                                  |
   +-----------------------------------------------------------*/

  /* Number of bytes per element (on average, if necessary) */
  sched_args.unit_size = sizeof(int);

  /* Iff nonzero, Creates one emit queue for each reduce task,
   * instead of per reduce thread. This improves time to emit 
   * if data is emitted in order, but can increase merge time. */
  sched_args.use_one_queue_per_task = 0;

  /* L1D cache size, in bytes */
  sched_args.L1_cache_size = 32*1024;

  /* Number of threads on which to execute map tasks (Default = 1 per processor) */
  sched_args.num_map_threads = 4;

  /* Number of threads on which to execute reduce tasks ((Default = 1 per processor) */
  sched_args.num_reduce_threads = 1;

  /* Number of threads on which to execute merge tasks ((Default = 1 per processor) */
  sched_args.num_merge_threads = 1;

  /* Number of processors to use ((Default (-1) = 1 per physical context)) */
  //  sched_args.num_procs = -1;
  sched_args.num_procs = 1; // Temporarily change to 2 by Shen Cheng

  /* Ratio of input data size to output data size */
  sched_args.key_match_factor = (float) (nElements); // reducing an array to a sum

  /*
   * RUN THE ALGORITHM
   */
#ifdef sparc
  starttime = gethrtime();
#endif
  if( map_reduce_scheduler( &sched_args ) < 0 ) {
    fatal("Scheduler had an error. Bailing out.\n");
  } 
#ifdef sparc
  endtime = gethrtime();
  mr_sumtime = endtime-starttime;
#endif

  keyval_t * p_pair = result.data;
  int * p_mr_key = (int*) p_pair[0].key;
  int * p_mr_sum = (int*) p_pair[0].val;
  int mrsum = *p_mr_sum;

  delete p_mr_key;
  delete p_mr_sum;

  cout << "MapReduce Sum: " << mrsum << endl;
  cout << "True Sum:      " << true_sum << endl;
#ifdef sparc
  cout << "MapReduce Time:   " << mr_sumtime << endl;
  cout << "Linear Scan Time: " << linearsumtime << endl;
  cout << "Speedup: " << ((double) linearsumtime) / ((double) mr_sumtime) << "x" << endl;
#else
  cout << "High-resolution timing not implemented on this platform." << endl;
#endif

  cout << endl;

  if( mrsum == true_sum ) {
    cout << "Correct Result." << endl;
  } else {
    cout << "+--------------------------------------+" << endl;
    cout << "| INCORRECT RESULT THE WORLD IS ENDING |" << endl;
    cout << "+--------------------------------------+" << endl;
  } 

  delete [] array;
  array = NULL;

  return 0;
} 
